﻿///////////////////////////////////////////////////////////////////////////////
// File:        Socket.cc
// Author:      671643387@qq.com
// Date:        2016年1月5日 下午5:43:27
// Description:
///////////////////////////////////////////////////////////////////////////////

#include <memory.h>
#if defined(__LINUX__)
#include <fcntl.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#endif
#include "Socket.hpp"
#include "SocketThread.hpp"
#include "SocketManager.hpp"

#define __LOCK(x)   fog::thread::LockGuard<fog::thread::Mutex> guard(x)
#define SOCKET_RECVBUFF_SIZE    0x1000

namespace fog
{
    namespace net
    {
        Socket::Socket(void)
            : id_(-1)
            , owner_(NULL)
            , actor_(NULL)
            , sock_(-1)
            , evRecv_(NULL)
            , evSend_(NULL)
            , evBase_(NULL)
            , recvBuff_(SOCKET_RECVBUFF_SIZE)
            , msgQueue_()
            , msgQueueLock_()
            , sendPending_(false)
            , sendPendingLOCK_()
        {
            memset(&addr_, 0, sizeof(addr_));
        }

        Socket::~Socket(void)
        {
            doClose();
        }

        bool Socket::open(evutil_socket_t s, SocketThread *t)
        {
            assert(s != -1);
            sock_ = s;
            evBase_ = t->evBase_;
            actor_ = t;

            do 
            {
                ev_socklen_t addrlen = sizeof(addr_);
                if (0 != ::getpeername(s, (struct sockaddr*)&addr_, &addrlen))
                    break;

                if (!Socket::setnonblocking(s))
                    break;

                evRecv_ = event_new(evBase_, s, 
                    EV_READ | EV_PERSIST, &Socket::onRecv, this);
                if (!evRecv_)
                    break;
                if (0 != event_add(evRecv_, NULL))
                    break;

                return true;
            } while (false);

            return false;
        }

        void Socket::close(void)
        {
            doClose();
            onClose();
            owner_->doCloseSocket(this);
        }

        std::string Socket::getPeerAddr(void)
        {
            return std::string(::inet_ntoa(addr_.sin_addr));
        }

        int Socket::getPeerPort(void)
        {
            return int(::ntohs(addr_.sin_port));
        }

        void Socket::send(const BYTE_t *buf, std::size_t len)
        {
            util::MessageBlockPtr mb(new util::MessageBlock(buf, len));
            if (!mb)
                close();
            else
            {
                util::MessageBlock *smb = NULL;
                {
                    __LOCK(msgQueueLock_);
                    msgQueue_.push_back(mb);
                    if (msgQueue_.size() == 1)
                        smb = mb.get();
                }
                trySend(smb);
            }
        }

        bool Socket::setnonblocking(evutil_socket_t s)
        {
#if defined(__WINDOWS__)
            ::u_long arg = 1;
            return (0 == ::ioctlsocket(s, FIONBIO, &arg));
#elif defined(__LINUX__)
            int flags = ::fcntl(s, F_GETFL, 0);
            flags |= O_NONBLOCK;
            return (-1 != ::fcntl(s, F_SETFL, flags));
#endif
        }

        void Socket::close(evutil_socket_t s)
        {
#if defined(__WINDOWS__)
            ::closesocket(s);
#elif defined(__LINUX__)
            ::close(s);
#endif
        }

        void Socket::onOpen(void)
        {

        }

        void Socket::onRecv(void)
        {

        }

        void Socket::onClose(void)
        {

        }

        int Socket::recv_net(BYTE_t *buf, int len)
        {
#if defined(__WINDOWS__)
            int ret = ::recv(sock_, (char*)buf, int(len), 0);
#elif defined(__LINUX__)
            int ret = ::recv(sock_, (char*)buf, len, 0);
#endif
            if (ret == SOCKET_ERROR)
            {
#if defined(__LINUX__)
                if (errno == EWOULDBLOCK)
                {
                    return SOCKET_ERROR_WOULDBLOCK;
                }
#elif defined(__WINDOWS__)
                int wsaError = WSAGetLastError();
                if (wsaError == WSAEWOULDBLOCK)
                {
                    return SOCKET_ERROR_WOULDBLOCK;
                }
#endif
            }
            return ret;
        }

        int Socket::send_net(const BYTE_t *buf, int len)
        {
#if defined(__WINDOWS__)
            int ret = ::send(sock_, (const char*)buf, int(len), 0);
#elif defined(__LINUX__)
            int ret = ::send(sock_, (const char*)buf, len, 0);
#endif
            if (ret == SOCKET_ERROR)
            {
#if defined(__LINUX__)
                if (errno == EWOULDBLOCK)
                {
                    return SOCKET_ERROR_WOULDBLOCK;
                }
#elif defined(__WINDOWS__)
                int wsaError = WSAGetLastError();
                if (wsaError == WSAEWOULDBLOCK)
                {
                    return SOCKET_ERROR_WOULDBLOCK;
                }
#endif
            }
            return ret;
        }

        bool Socket::notifySend(void)
        {
            assert(evBase_);
            if (!evSend_)
            {
                evSend_ = event_new(evBase_, sock_,
                    EV_WRITE, &Socket::onSend, this);
            }
            return (evSend_ && (0 == event_add(evSend_, NULL)));
        }

        void Socket::trySend(util::MessageBlock *mb)
        {
            if (mb == NULL)
            {
                __LOCK(sendPendingLOCK_);
                if (sendPending_)
                    return;
                {
                    __LOCK(msgQueueLock_);
                    if (!msgQueue_.empty())
                        mb = msgQueue_.front().get();
                }

                if (mb)
                    sendPending_ = true;
                else
                    return;
            }
            else
            {
                __LOCK(sendPendingLOCK_);
                if (sendPending_)
                    return;
                sendPending_ = true;
            }

            do 
            {
                if (!notifySend())
                    break;

                int nSend = send_net(mb->contents(), mb->size());
                if (nSend == SOCKET_ERROR)
                    break;

                if (nSend > 0)
                    mb->rpos(nSend);

                if (mb->size() <= 0)
                {
                    __LOCK(msgQueueLock_);
                    msgQueue_.pop_front();
                }

                return;
            } while (false);

            close();
        }

        void Socket::doClose(void)
        {
            if (evRecv_)
            {
                event_free(evRecv_);
                evRecv_ = NULL;
            }

            if (evSend_)
            {
                event_free(evSend_);
                evSend_ = NULL;
            }

            if (sock_ != -1)
            {
                Socket::close(sock_);
                sock_ = -1;
            }
        }

        void Socket::doRecv(void)
        {
            bool fOk = true;
            while (recvBuff_.space() > 0)
            {
                BYTE_t buff[0x1000];
                std::size_t bytesToRecv = __MIN_(recvBuff_.space(), sizeof(buff));
                int nRead = recv_net(buff, int(bytesToRecv));
                if (nRead > 0)
                    recvBuff_.write(buff, nRead);
                else if (nRead == SOCKET_ERROR_WOULDBLOCK)
                    break;
                else
                {
                    fOk = false;
                    break;
                }
            }
            if (!fOk)
                close();
            else
                onRecv();
        }

        void Socket::onRecv(evutil_socket_t sock, short event, void* arg)
        {
            assert(arg);
            Socket *pThis = static_cast<Socket*>(arg);
            pThis->doRecv();
        }

        void Socket::doSend(void)
        {
            {
                __LOCK(sendPendingLOCK_);
                sendPending_ = false;
            }
            trySend(NULL);
        }

        void Socket::onSend(evutil_socket_t sock, short event, void* arg)
        {
            assert(arg);
            Socket *pThis = static_cast<Socket*>(arg);
            pThis->doSend();
        }
    }
}